package com.behavioranalysis.flinkprogram.flink.session;

import com.alibaba.fastjson.JSONObject;
import com.behavioranalysis.flinkprogram.conf.ConfigurationManager;
import com.behavioranalysis.flinkprogram.constant.Constants;
import com.behavioranalysis.flinkprogram.dao.ISessionAggrStatDAO;
import com.behavioranalysis.flinkprogram.dao.ISessionRandomExtractDAO;
import com.behavioranalysis.flinkprogram.dao.ITaskDAO;
import com.behavioranalysis.flinkprogram.dao.factory.DAOFactory;
import com.behavioranalysis.flinkprogram.dao.output.JDBCOutputFormat;
import com.behavioranalysis.flinkprogram.domain.SessionAggrStat;
import com.behavioranalysis.flinkprogram.domain.Task;
import com.behavioranalysis.flinkprogram.util.*;
import org.apache.commons.lang3.mutable.MutableLong;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.FlatMapIterator;
import org.apache.flink.api.java.tuple.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 用户session分析
 * <p>
 * user_info
 * "user_id","username","name","age","professional","city","sex"
 * LONG      STRING    STRING  INT    STRING       STRING  STRING
 * <p>
 * user_visit_action
 * <p>
 * "date",                  Types.STRING,
 * "user_id",               Types.LONG,
 * "session_id",            Types.STRING,
 * "page_id",               Types.LONG,
 * "action_time",           Types.STRING,
 * "search_keyword",        Types.STRING,
 * "click_category_id",     Types.LONG,
 * "click_product_id",      Types.LONG,
 * "order_category_ids",    Types.STRING,
 * "order_product_ids",     Types.STRING,
 * "pay_category_ids",      Types.STRING,
 * "pay_product_ids",       Types.STRING,
 * "city_id"                Types.LONG
 */
public class UserVisitSessionAnalyze2 {
    public static void main(String[] args) throws Exception {
        // 构建flink上下文
        ExecutionEnvironment env = getEnv();
        // 设置并行度
        // 并行度暂设为1，后面排序要用到
        env.setParallelism(1);

        DataSet<String> actionDataSet = getActionDataSet(env);
        DataSet<String> userInfoDataSet = getUserInfoDataSet(env);


        DataSet<Row> userInfoRow = userInfoDataSet.map(
                (MapFunction<String, Row>) value -> {
                    String[] splitedStr = value.split("\u0001");
                    Row row = Row.of(splitedStr[0], splitedStr[1], splitedStr[2],
                            splitedStr[3], splitedStr[4], splitedStr[5], splitedStr[6]);
                    return row;
                }
        );
        System.out.println("=====================userInfo Start======================");
        userInfoRow.print();
        System.out.println("=====================userInfo End======================");

        // 映射成<userid,Row>的格式
        DataSet<Tuple2<Long, Row>> userid2userInfo = userInfoRow.map(
                (MapFunction<Row, Tuple2<Long, Row>>) row ->
                        new Tuple2<>(Long.valueOf(String.valueOf(row.getField(0))), row)
        );

        DataSet<Row> actionRow = actionDataSet.map(
                (MapFunction<String, Row>) value -> {
                    String[] splitedStr = value.split("\u0001");
                    Row row = Row.of(splitedStr[0], splitedStr[1], splitedStr[2],
                            splitedStr[3], splitedStr[4], splitedStr[5], splitedStr[6],
                            splitedStr[7], splitedStr[8], splitedStr[9], splitedStr[10]
                            , splitedStr[11], splitedStr[12]);
                    return row;
                }
        );

        System.out.println("=====================action Start======================");
        actionRow.print();
        System.out.println("=====================action End======================");

        // 创建需要使用的DAO组件
        ITaskDAO taskDAO = DAOFactory.getTaskDAO();

        // 那么就首先的查询出来指定的任务，并获取任务的查询参数
        long taskid = ParamUtils.getTaskIdFromArgs(env, args, Constants.FLINK_LOCAL_TASKID_SESSION);
        Task task = taskDAO.findById(taskid);
        if (task == null) {
            System.out.println(new Date() + ": cannot find this task with id [" + taskid + "].");
            return;
        }
        System.out.println("=====================timingTask Start======================");
        System.out.println(task.toString());
        System.out.println("=====================timingTask End======================");

        JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());
        // 将actionRow进行过滤得到指定日期内的行为数据
        // filter时会用到外部的日期数据，需要向filter函数传递参数
        // 这里我们使用ExecutionConfig向Function传递参数，另一种是利用构造方法传参
        Configuration configuration = new Configuration();

        /**
         * 一 、 查询出来指定日期内的行为数据，并将其映射为<sessionid, row>的tuple
         */
        System.out.println("=====================一、 Start======================");
        DataSet<Row> actionDateRow = getActionRowByDateRange(actionRow, taskid, taskParam, configuration);
        // TODO 此DataSet sessionid2action后面会用到
        DataSet<Tuple2<String, Row>> sessionid2action = getSessionid2Action(actionDateRow);
        System.out.println("=====================一、 End======================");

        /**
         * 二、将session粒度的数据与user数据进行join
         * 得到<sessionid,(sessionid,searchKeywords,clickCategoryIds,visitLength,stepLength,startTime,age,professional,city,sex)>格式的tuple
         */
        System.out.println("=====================二、 Start======================");
        DataSet<Tuple2<String, String>> sessionid2FullAggrInfo = aggregateBySession(sessionid2action, userid2userInfo);
        System.out.println("=====================二、 End======================");

        /**
         * 三、接着，就要针对session粒度的聚合数据，按照使用者指定的筛选参数进行数据过滤
         * 相当于我们自己编写的算子，是要访问外面的任务参数对象的
         */
        System.out.println("=====================三、 Start======================");
        // TODO 此dataset filterSessionid2AggrInfo后面会用到
        DataSet<Tuple2<String, String>> filterSessionid2AggrInfo = filterSessionAndAggrStat(sessionid2FullAggrInfo, taskParam, configuration);
        System.out.println("=====================三、 End======================");

        /**
         * 四、获取通过筛选的session的访问明细数据
         */
        System.out.println("=====================四、 Start======================");
        // TODO 此dataSet sessionid2detail后面会用到
        DataSet<Tuple2<String, Row>> sessionid2detail = filterSessionid2AggrInfo
                .join(sessionid2action)
                .where(0)
                .equalTo(0)
                .projectFirst(0).projectSecond(1);
        System.out.println("=====================四、 End======================");
        //sessionid2detail.print();
        //System.out.println("filterSessionid2AggrInfo's count: " + filterSessionid2AggrInfo.count());
        //System.out.println("sessionid2detail's count: " + sessionid2detail.count());

        /**
         * 五、随机抽取session
         */
        System.out.println("=====================五、 Start======================");
        randomExtractSession(filterSessionid2AggrInfo, sessionid2detail, configuration, env);
        System.out.println("=====================五、 End======================");

        /**
         * 六、获取top10热门品类
         */
        System.out.println("=====================六、 Start======================");
        // TODO 此dataset top10Category后面会用到
        DataSet<Tuple5<Long, Long, Long, Long, Long>> top10Category = getTop10Category(sessionid2detail, configuration);
        System.out.println("=====================六、 End======================");

        /**
         * 七、获取top10活跃session
         */
        System.out.println("=====================七、 Start======================");
        getTop10Session(sessionid2detail, top10Category);
        System.out.println("=====================七、 End======================");

        /**
         * 有sink操作必须调用execute方法来触发，第一次好像是这样！！！
         *
         * 累加器的整个结果将存储在JobExecutionResult从execute()执行环境的方法返回的对象中（当前这仅在执行等待作业完成时才有效）。
         *
         * eg. myJobExecutionResult.getAccumulatorResult("num-lines")
         *
         *    * 六、计算出各个范围的session占比，并写入MySQL
         *
         */
        JobExecutionResult sessionJobExecutionResult = env.execute("UserVisitSessionAnalyze");
        ConcurrentSkipListMap<String, MutableLong> sessionAggrStatMap = sessionJobExecutionResult.getAccumulatorResult("sessionAggrStat");
        //System.out.println(sessionAggrStatMap);
        System.out.println("=====================八、 Start======================");
        calculateAndPersistAggrStat(sessionAggrStatMap, taskid);
        System.out.println("=====================八、 End======================");
    }

    private static DataSet<String> getUserInfoDataSet(ExecutionEnvironment env) {
        DataSet<String> userInfoDataSet;
        if (ConfigurationManager.getBoolean(Constants.FLINK_LOCAL)) {
            //actionDataSet = env.readTextFile("/home/yang/文档/毕业设计/user_visit_action.txt");
            //userInfoDataSet = env.readTextFile("/home/yang/文档/毕业设计/user_info.txt");
            userInfoDataSet = env.readTextFile("hdfs://192.168.43.111:9000/visit-logs/user_info.txt");
        } else {
            userInfoDataSet = env.readTextFile("hdfs://192.168.43.111:9000/visit-logs/user_info.txt");
        }
        return userInfoDataSet;
    }

    private static DataSet<String> getActionDataSet(ExecutionEnvironment env) {
        DataSet<String> actionDataSet;
        if (ConfigurationManager.getBoolean(Constants.FLINK_LOCAL)) {
            //actionDataSet = env.readTextFile("/home/yang/文档/毕业设计/user_visit_action.txt");
            //userInfoDataSet = env.readTextFile("/home/yang/文档/毕业设计/user_info.txt");
            actionDataSet = env.readTextFile("hdfs://192.168.43.111:9000/visit-logs/user_visit_action.txt");
        } else {
            actionDataSet = env.readTextFile("hdfs://192.168.43.111:9000/visit-logs/user_visit_action.txt");
        }
        return actionDataSet;
    }

    private static ExecutionEnvironment getEnv() {
        ExecutionEnvironment env;
        if (ConfigurationManager.getBoolean(Constants.FLINK_LOCAL)) {
            env = ExecutionEnvironment.createLocalEnvironment();
        } else {
            env = ExecutionEnvironment.getExecutionEnvironment();
        }
        return env;
    }

    /**
     * 获取指定日期类的行为数据
     *
     * @param actionRow
     * @param taskid
     * @param taskParam
     * @param configuration
     * @return
     * @throws ParseException
     */
    private static DataSet<Row> getActionRowByDateRange(DataSet<Row> actionRow, long taskid, JSONObject taskParam, Configuration configuration) throws ParseException {
        String startDate = ParamUtils.getParam(taskParam, Constants.PARAM_START_DATE).trim();
        String endDate = ParamUtils.getParam(taskParam, Constants.PARAM_END_DATE).trim();
        long startDateLong = new SimpleDateFormat("yyyy-MM-dd").parse(startDate).getTime();
        long endDateLong = new SimpleDateFormat("yyyy-MM-dd").parse(endDate).getTime();

        configuration.setLong("startDateLong", startDateLong);
        configuration.setLong("endDateLong", endDateLong);
        configuration.setLong("taskid", taskid);
        DataSet<Row> actionDateRow = actionRow.filter(
                new RichFilterFunction<Row>() {
                    private long startTimeLong;
                    private long endTimeLong;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 获取Configuration传递过来的参数
                        startTimeLong = parameters.getLong("startDateLong", 0L);
                        endTimeLong = parameters.getLong("endDateLong", 0L);
                    }

                    @Override
                    public boolean filter(Row value) throws Exception {
                        String date = value.getField(0).toString().trim();
                        long dateLong = new SimpleDateFormat("yyyy-MM-dd").parse(date).getTime();

                        long dateStartLong = dateLong - startTimeLong;
                        long endDateLong = endTimeLong - dateLong;

                        // startDateKey <= date <= endDateKey
                        if (dateStartLong >= 0 && endDateLong >= 0) {
                            return true;
                        }
                        return false;
                    }
                }
        ).withParameters(configuration);
        return actionDateRow;
    }

    /**
     * 将指定日期类的行为数据映射为<sessionid, row>的tuple
     *
     * @param actionDateRow
     * @return
     */
    private static DataSet<Tuple2<String, Row>> getSessionid2Action(DataSet<Row> actionDateRow) {
        DataSet<Tuple2<String, Row>> sessionid2action = actionDateRow.map(
                (MapFunction<Row, Tuple2<String, Row>>) row ->
                        new Tuple2<>(String.valueOf(row.getField(2)), row)
        );
        return sessionid2action;
    }

    /**
     * 将session粒度的数据与user数据进行join
     *
     * @param sessionid2action
     * @param userid2userInfo
     * @return
     */
    private static DataSet<Tuple2<String, String>> aggregateBySession(DataSet<Tuple2<String, Row>> sessionid2action, DataSet<Tuple2<Long, Row>> userid2userInfo) {
        //1. 对行为数据按session粒度进行分组
        //   并对每一个session分组进行聚合，将session中所有的搜索词和点击品类都聚合起来
        //   到此为止，获取的数据格式，如下：
        //   <userid, partAggrInfo(sessionid,searchKeywords,clickCategoryIds)
        DataSet<Tuple2<Long, String>> userid2PartAggrInfo = sessionid2action
                .groupBy(0)
                .combineGroup((GroupCombineFunction<Tuple2<String, Row>, Tuple2<Long, String>>) (values, out) -> {
                    String sessionid = "";
                    StringBuffer searchKeywordsBuffer = new StringBuffer("");
                    StringBuffer clickCategoryIdsBuffer = new StringBuffer("");
                    Long userid = null;
                    // session的起始和结束时间
                    Date startTime = null;
                    Date endTime = null;
                    // session的访问步长
                    int stepLength = 0;
                    // 遍历session所有的访问行为
                    for (Tuple2<String, Row> tuple2 : values) {
                        if ("".equals(sessionid)) {
                            sessionid = tuple2.f0;
                        }
                        Row row = tuple2.f1;
                        if (userid == null) {
                            userid = Long.valueOf(String.valueOf(row.getField(1)));
                        }
                        String searchKeyword = String.valueOf(row.getField(5));
                        String clickCategoryId = String.valueOf(row.getField(6));
                        // 并不是每一行访问行为都有searchKeyword和clickCategoryId两个字段的
                        // 是否将搜索词和点击品类id拼接到字符串中去
                        // 首先要满足：不能是null值，其次：之前的字符串中还没有该搜索词和该点击品类id
                        if (StringUtils.isNotEmpty(searchKeyword)) {
                            if (!searchKeywordsBuffer.toString().contains(searchKeyword)) {
                                searchKeywordsBuffer.append(searchKeyword + ",");
                            }
                        }
                        if (StringUtils.isNotEmpty(clickCategoryId)) {
                            if (!clickCategoryIdsBuffer.toString().contains(clickCategoryId)) {
                                clickCategoryIdsBuffer.append(clickCategoryId + ",");
                            }
                        }
                        // 计算session开始和结束时间
                        Date actionTime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(String.valueOf(row.getField(4)));
                        if (startTime == null) {
                            startTime = actionTime;
                        }
                        if (endTime == null) {
                            endTime = actionTime;
                        }
                        if (actionTime.before(startTime)) {
                            startTime = actionTime;
                        }
                        if (actionTime.after(endTime)) {
                            endTime = actionTime;
                        }
                        // 计算session的访问步长
                        stepLength++;
                    }
                    String searchKeywords = StringUtils.trimComma(searchKeywordsBuffer.toString());
                    String clickCategoryIds = StringUtils.trimComma(clickCategoryIdsBuffer.toString());
                    // 计算session访问时长(秒)
                    long visitLength = (endTime.getTime() - startTime.getTime()) / 1000;
                    // 要join userinfo,所以先返回<userid,partAggrInfo>,后面再将返回的Tuple的key设置为sessionid
                    // 聚合数据，使用key=value|key=value的格式进行拼接
                    String partAggrInfo = Constants.FIELD_SESSION_ID + "=" + sessionid + "|"
                            + Constants.FIELD_SEARCH_KEYWORDS + "=" + searchKeywords + "|"
                            + Constants.FIELD_CLICK_CATEGORY_IDS + "=" + clickCategoryIds + "|"
                            + Constants.FIELD_VISIT_LENGTH + "=" + visitLength + "|"
                            + Constants.FIELD_STEP_LENGTH + "=" + stepLength + "|"
                            + Constants.FIELD_START_TIME + "=" + DateUtils.formatTime(startTime);
                    out.collect(new Tuple2<>(userid, partAggrInfo));
                });

        // 2. 将session粒度聚合数据，与用户信息进行join,并且返回<sessionid,fullAggrInfo>格式的数据
        DataSet<Tuple2<String, String>> sessionid2FullAggrInfo = userid2PartAggrInfo
                .join(userid2userInfo)
                .where(0)
                .equalTo(0)
                .with((JoinFunction<Tuple2<Long, String>, Tuple2<Long, Row>, Tuple2<String, String>>) (first, second) -> {
                    String partAggrInfo = first.f1;
                    String sessionid = StringUtils.getFieldFromConcatString(
                            partAggrInfo, "\\|", Constants.FIELD_SESSION_ID);
                    Row userInfoRow = second.f1;
                    int age = Integer.valueOf(String.valueOf(userInfoRow.getField(3)));
                    String professional = String.valueOf(userInfoRow.getField(4));
                    String city = String.valueOf(userInfoRow.getField(5));
                    String sex = String.valueOf(userInfoRow.getField(6));
                    String fullAggrInfo = partAggrInfo + "|"
                            + Constants.FIELD_AGE + "=" + age + "|"
                            + Constants.FIELD_PROFESSIONAL + "=" + professional + "|"
                            + Constants.FIELD_CITY + "=" + city + "|"
                            + Constants.FIELD_SEX + "=" + sex;
                    return new Tuple2<>(sessionid, fullAggrInfo);
                });
        return sessionid2FullAggrInfo;
    }

    /**
     * 对session粒度的聚合数据，按照使用者指定的筛选参数进行数据过滤
     *
     * @param sessionid2FullAggrInfo
     * @param taskParam
     * @param configuration
     * @return
     */
    private static DataSet<Tuple2<String, String>> filterSessionAndAggrStat(DataSet<Tuple2<String, String>> sessionid2FullAggrInfo, JSONObject taskParam, Configuration configuration) {
        String startAge = ParamUtils.getParam(taskParam, Constants.PARAM_START_AGE);
        String endAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AGE);
        String professionals = ParamUtils.getParam(taskParam, Constants.PARAM_PROFESSIONALS);
        String cities = ParamUtils.getParam(taskParam, Constants.PARAM_CITIES);
        String sex = ParamUtils.getParam(taskParam, Constants.PARAM_SEX);
        String keywords = ParamUtils.getParam(taskParam, Constants.PARAM_KEYWORDS);
        String categoryIds = ParamUtils.getParam(taskParam, Constants.PARAM_CATEGORY_IDS);

        String _parameter =
                (startAge != null ? Constants.PARAM_START_AGE + "=" + startAge + "|" : "")
                        + (endAge != null ? Constants.PARAM_END_AGE + "=" + endAge + "|" : "")
                        + (professionals != null ? Constants.PARAM_PROFESSIONALS + "=" + professionals + "|" : "")
                        + (cities != null ? Constants.PARAM_CITIES + "=" + cities + "|" : "")
                        + (sex != null ? Constants.PARAM_SEX + "=" + sex + "|" : "")
                        + (keywords != null ? Constants.PARAM_KEYWORDS + "=" + keywords + "|" : "")
                        + (categoryIds != null ? Constants.PARAM_CATEGORY_IDS + "=" + categoryIds : "");

        if (_parameter.endsWith("\\|")) {
            _parameter = _parameter.substring(0, _parameter.length() - 1);
        }
        configuration.setString("_parameter", _parameter);
        DataSet<Tuple2<String, String>> filterSessionid2AggrInfo = sessionid2FullAggrInfo
                .filter(new RichFilterFunction<Tuple2<String, String>>() {
                    private String parameter;
                    // 使用累加器
                    private final SessionAggrStatAccumulator sessionAggrStatAccumulator = new SessionAggrStatAccumulator();

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        this.getRuntimeContext().addAccumulator("sessionAggrStat", sessionAggrStatAccumulator);
                        parameter = parameters.getString("_parameter", "");
                    }

                    @Override
                    public boolean filter(Tuple2<String, String> value) throws Exception {
                        // 首先从tuple中，获取聚合数据
                        String aggrInfo = value.f1;
                        // 接着，依次按照筛选条件进行过滤
                        // 按照年龄范围进行过滤(startAge、endAge)
                        if (!ValidUtils.between(aggrInfo, Constants.FIELD_AGE, parameter, Constants.PARAM_START_AGE, Constants.PARAM_END_AGE)) {
                            return false;
                        }
                        // 按照职业范围进行过滤（professionals）
                        if (!ValidUtils.in(aggrInfo, Constants.FIELD_CITY, parameter, Constants.PARAM_CITIES)) {
                            return false;
                        }
                        // 按照性别进行过滤
                        if (!ValidUtils.equal(aggrInfo, Constants.FIELD_SEX, parameter, Constants.PARAM_SEX)) {
                            return false;
                        }
                        // 按照搜索词进行过滤，可能有多个搜索词，只要有任何一个与筛选天剑中相同，即通过
                        if (!ValidUtils.in(aggrInfo, Constants.FIELD_SEARCH_KEYWORDS, parameter, Constants.PARAM_KEYWORDS)) {
                            return false;
                        }
                        // 按照点击品类id进行过滤
                        if (!ValidUtils.in(aggrInfo, Constants.FIELD_CLICK_CATEGORY_IDS, parameter, Constants.PARAM_CATEGORY_IDS)) {
                            return false;
                        }
                        // 如果经过了之前的多个过滤条件之后，程序能够走到这里
                        // 那么说明，该session是通过了用户指定的筛选条件的，也就是需要保留的session
                        // 那么就要对session的访问时长和访问步长，进行统计
                        // 进行相应的累加计数

                        // 主要走到这一步，那么就是需要计数的session
                        this.sessionAggrStatAccumulator.add(Constants.SESSION_COUNT);

                        // 计算出session的访问时长和访问步长的范围，并进行相应的累加
                        long visitLength = Long.valueOf(StringUtils.getFieldFromConcatString(
                                aggrInfo, "\\|", Constants.FIELD_VISIT_LENGTH));
                        long stepLength = Long.valueOf(StringUtils.getFieldFromConcatString(
                                aggrInfo, "\\|", Constants.FIELD_STEP_LENGTH));
                        calculateVisionLength(visitLength);
                        calculateStepLength(stepLength);

                        return true;
                    }

                    /**
                     * 计算访问时长范围
                     * @param visitLength
                     */
                    private void calculateVisionLength(long visitLength) {
                        if (visitLength >= 1 && visitLength <= 3) {
                            this.sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1s_3s);
                        } else if (visitLength >= 4 && visitLength <= 6) {
                            this.sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_4s_6s);
                        } else if (visitLength >= 7 && visitLength <= 9) {
                            this.sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_7s_9s);
                        } else if (visitLength >= 10 && visitLength <= 30) {
                            this.sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10s_30s);
                        } else if (visitLength > 30 && visitLength <= 60) {
                            this.sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30s_60s);
                        } else if (visitLength > 60 && visitLength <= 180) {
                            this.sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1m_3m);
                        } else if (visitLength > 180 && visitLength <= 600) {
                            this.sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_3m_10m);
                        } else if (visitLength > 600 && visitLength <= 1800) {
                            this.sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10m_30m);
                        } else if (visitLength > 1800) {
                            this.sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30m);
                        }
                    }

                    /**
                     * 计算访问步长范围
                     * @param stepLength
                     */
                    private void calculateStepLength(long stepLength) {
                        if (stepLength >= 1 && stepLength <= 3) {
                            this.sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_1_3);
                        } else if (stepLength >= 4 && stepLength <= 6) {
                            this.sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_4_6);
                        } else if (stepLength >= 7 && stepLength <= 9) {
                            this.sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_7_9);
                        } else if (stepLength >= 10 && stepLength <= 30) {
                            this.sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_10_30);
                        } else if (stepLength > 30 && stepLength <= 60) {
                            this.sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_30_60);
                        } else if (stepLength > 60) {
                            this.sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_60);
                        }
                    }
                }).withParameters(configuration);
        return filterSessionid2AggrInfo;
    }

    /**
     * 随机抽取session
     *
     * @param filterSessionid2AggrInfo
     * @param sessionid2detail
     * @param configuration
     * @param env
     */
    private static void randomExtractSession(DataSet<Tuple2<String, String>> filterSessionid2AggrInfo, DataSet<Tuple2<String, Row>> sessionid2detail, Configuration configuration, ExecutionEnvironment env) throws Exception {
        // 第一步，计算每天每小时的session数量， <yyyy-MM-dd_HH,aggrInfo>格式的Tuple
        DataSet<Tuple2<String, String>> time2session = filterSessionid2AggrInfo.map(
                (MapFunction<Tuple2<String, String>, Tuple2<String, String>>) value -> {
                    String aggrInfo = value.f1;
                    String startTime = StringUtils.getFieldFromConcatString(aggrInfo, "\\|", Constants.FIELD_START_TIME);
                    String dateHour = DateUtils.getDateHour(startTime);
                    return new Tuple2<>(dateHour, aggrInfo);
                }
        );
        /**
         * 思考一下：这里我们不要着急写大量的代码，做项目的时候，一定要用脑子多思考
         *
         * 每天每小时的session数量，然后计算出每天每小时的session抽取索引，遍历每天每小时session
         * 首先抽取出的session的聚合数据，写入session_random_extract表
         * 所以第一个RDD的value，应该是session聚合数据
         */

        // 得到每天每小时的session数量
        DataSet<Tuple2<String, Long>> time2count = time2session
                .map((MapFunction<Tuple2<String, String>, Tuple2<String, Long>>) value -> new Tuple2<>(value.f0, 1L))
                .groupBy(0)
                .sum(1);
        List<Tuple2<String, Long>> countList = time2count.collect();
        Map<String, Long> countMap = new HashMap<>();
        for (Tuple2<String, Long> tuple2 : countList) {
            countMap.put(tuple2.f0, tuple2.f1);
        }

        // 第二步，使用按时间比例随机抽取算法，计算出每天每小时要抽取session的索引
        // 将<yyyy-MM-dd_HH, count>格式的map，转换为<yyyy-MM-dd, <HH, count>>的格式
        Map<String, Map<String, Long>> dateHourCountMap = new HashMap<>();
        for (Map.Entry<String, Long> countEntry : countMap.entrySet()) {
            String dateHour = countEntry.getKey();
            String date = dateHour.split("_")[0];
            String hour = dateHour.split("_")[1];
            long count = countEntry.getValue();
            Map<String, Long> hourCountMap = dateHourCountMap.get(date);
            if (hourCountMap == null) {
                hourCountMap = new HashMap<>();
                dateHourCountMap.put(date, hourCountMap);
            }
            hourCountMap.put(hour, count);
        }

        // 开始实现 按时间比例随机抽取算法
        // 总共要抽取100个session，先按照天数，进行平分
        int extractNumberPerDay = 100 / dateHourCountMap.size();
        Map<String, Map<String, List<Integer>>> dateHourExtractMap = new HashMap<>();
        Random random = new Random();
        for (Map.Entry<String, Map<String, Long>> dateHourCountEntry : dateHourCountMap.entrySet()) {
            String date = dateHourCountEntry.getKey();
            Map<String, Long> hourCountMap = dateHourCountEntry.getValue();
            // 计算出这一天的session总数
            long sessionCount = 0L;
            for (long hourCount : hourCountMap.values()) {
                sessionCount += hourCount;
            }
            Map<String, List<Integer>> hourExtractMap = dateHourExtractMap.get(date);
            if (hourExtractMap == null) {
                hourExtractMap = new HashMap<>();
                dateHourExtractMap.put(date, hourExtractMap);
            }
            //遍历每个小时
            for (Map.Entry<String, Long> hourCountEntry : hourCountMap.entrySet()) {
                String hour = hourCountEntry.getKey();
                long count = hourCountEntry.getValue();
                // 计算每个小时的session数量，占据当天总session数量的比例，直接乘以每天要抽取的数量
                // 就可以计算出，当前小时需要抽取的session数量
                int hourExtractNumber = (int) (((double) count / (double) sessionCount) * extractNumberPerDay);
                if (hourExtractNumber > count) {
                    hourExtractNumber = (int) count;
                }
                // 先获取当前小时的存放随机数的list
                List<Integer> extractIndexList = hourExtractMap.get(hour);
                if (extractIndexList == null) {
                    extractIndexList = new ArrayList<>();
                    hourExtractMap.put(hour, extractIndexList);
                }
                // 生成上面计算出来的数量的随机数
                for (int i = 0; i < hourExtractNumber; i++) {
                    int extractIndex = random.nextInt((int) count);
                    while (extractIndexList.contains(extractIndex)) {
                        extractIndex = random.nextInt((int) count);
                    }
                    extractIndexList.add(extractIndex);
                }
            }
        }
        //System.out.println(dateHourExtractMap);

        Map<String, Map<String, Map<String, List<Integer>>>> transferMap = new HashMap<>();
        transferMap.put("transferMap", dateHourExtractMap);
        DataSet<Map<String, Map<String, List<Integer>>>> dateHourExtractMapDataSet = env.fromCollection(transferMap.values());
        //dateHourExtractMapDataSet.print();

        // TODO 可以使用 fastutil

        // 第三步，遍历每天每小时的session，然后根据随机索引进行抽取
        // 将随机抽取索引dateHourExtractMap做成广播变量
        DataSet<Tuple5<Long, String, String, String, String>> sessionRandomExtract = time2session
                .groupBy(0)
                .combineGroup(new RichGroupCombineFunction<Tuple2<String, String>, Tuple5<Long, String, String, String, String>>() {
                    private Map<String, Map<String, List<Integer>>> dateHourExtractMap;
                    private long taskid;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // Access the broadcast DataSet as a Collection
                        Collection<Map<String, Map<String, List<Integer>>>> dateHourExtractMapColl = getRuntimeContext().getBroadcastVariable("dateHourExtractMapDataSet");
                        dateHourExtractMap = ((List<Map<String, Map<String, List<Integer>>>>) dateHourExtractMapColl).get(0);
                        taskid = parameters.getLong("taskid", 0L);
                    }

                    @Override
                    public void combine(Iterable<Tuple2<String, String>> values, Collector<Tuple5<Long, String, String, String, String>> out) throws Exception {
                        Iterator<Tuple2<String, String>> iterator = values.iterator();
                        String dateHour = "";
                        String date = "";
                        String hour = "";
                        List<Integer> extractIndexList = new ArrayList<>();
                        ISessionRandomExtractDAO sessionRandomExtractDAO = DAOFactory.getSessionRandomExtractDAO();
                        int index = 0;
                        while (iterator.hasNext()) {
                            Tuple2<String, String> tuple2 = iterator.next();
                            String sessionAggrInfo = tuple2.f1;
                            if ("".equals(dateHour)) {
                                dateHour = tuple2.f0;
                                date = dateHour.split("_")[0];
                                hour = dateHour.split("_")[1];
                                extractIndexList = dateHourExtractMap.get(date).get(hour);
                            }
                            if (extractIndexList.contains(index)) {
                                String sessionid = StringUtils.getFieldFromConcatString(sessionAggrInfo, "\\|", Constants.FIELD_SESSION_ID);
                                String startTime = StringUtils.getFieldFromConcatString(
                                        sessionAggrInfo, "\\|", Constants.FIELD_START_TIME);
                                String searchKeywords = StringUtils.getFieldFromConcatString(
                                        sessionAggrInfo, "\\|", Constants.FIELD_SEARCH_KEYWORDS);
                                String clickCategoryIds = StringUtils.getFieldFromConcatString(
                                        sessionAggrInfo, "\\|", Constants.FIELD_CLICK_CATEGORY_IDS);
                                // startAge != null ? Constants.PARAM_START_AGE + "=" + startAge + "|" : ""
                                // 返回Taskid, Sessionid, StartTime, SearchKeywords, ClickCategoryIds
                                // 格式化 null 值
                                out.collect(new Tuple5<>(taskid, sessionid, startTime,
                                        searchKeywords != null ? searchKeywords : "无",
                                        clickCategoryIds != null ? clickCategoryIds : "无"));
                            }
                            index++;
                        }
                    }
                })
                .withBroadcastSet(dateHourExtractMapDataSet, "dateHourExtractMapDataSet")
                .withParameters(configuration);
        //sessionRandomExtract.print();
        //System.out.println(sessionRandomExtract.count());

        // 第四步，将sessionRandomExtract sink到mysql中
        // 输出到数据库可能为null的字段都必须格式化，否则会报空指针异常

        if (ConfigurationManager.getBoolean(Constants.FLINK_LOCAL)) {
            sessionRandomExtract.output(
                    JDBCOutputFormat
                            .buildJDBCOutputFormat()
                            .setDrivername(ConfigurationManager.getProperty(Constants.JDBC_DRIVER))
                            .setDBUrl(ConfigurationManager.getProperty(Constants.JDBC_URL))
                            .setUsername(ConfigurationManager.getProperty(Constants.JDBC_USER))
                            .setPassword(ConfigurationManager.getProperty(Constants.JDBC_PASSWORD))
                            .setQuery(ConfigurationManager.getProperty(Constants.SQL_SESSION_RANDOM_EXTRACT))
                            .finish()
            );
        } else {
            sessionRandomExtract.output(
                    JDBCOutputFormat
                            .buildJDBCOutputFormat()
                            .setDrivername(ConfigurationManager.getProperty(Constants.JDBC_DRIVER))
                            .setDBUrl(ConfigurationManager.getProperty(Constants.JDBC_URL_PROD))
                            .setUsername(ConfigurationManager.getProperty(Constants.JDBC_USER_PROD))
                            .setPassword(ConfigurationManager.getProperty(Constants.JDBC_PASSWORD_PROD))
                            .setQuery(ConfigurationManager.getProperty(Constants.SQL_SESSION_RANDOM_EXTRACT))
                            .finish()
            );
        }


        // 第五步，使用抽取出来的session （sessionid），去join它们的访问行为明细数据，并写入数据库
        // DataSet<Tuple5<Long, String, String, String, String>> sessionRandomExtract
        // join
        // DataSet<Tuple2<String, Row>> sessionid2detail
        DataSet<Tuple12<Long, Long, String, Long, String, String, Long, Long, String, String, String, String>> extractSessionDetail = sessionRandomExtract
                .join(sessionid2detail)
                .where(1)
                .equalTo(0)
                .with((JoinFunction<Tuple5<Long, String, String, String, String>, Tuple2<String, Row>, Tuple12<Long, Long, String, Long, String, String, Long, Long, String, String, String, String>>) (first, second) -> {
                    long taskid = first.f0;
                    Row row = second.f1;
                    // 此处返回的数据等会要通过JDBCOutputFormat sink到mysql中，
                    // 所以其中不能出现null值，否则会报空指针异常
                    // Taskid Userid Sessionid Pageid ActionTime SearchKeyword ClickCategoryId ClickProductId OrderCategoryIds OrderProductIds PayCategoryIds PayProductIds
                    return new Tuple12<>(
                            taskid,
                            Long.valueOf(String.valueOf(row.getField(1))),
                            first.f1,
                            Long.valueOf(String.valueOf(row.getField(3))),
                            String.valueOf(row.getField(4)),
                            StringUtils.isNotEmpty(String.valueOf(row.getField(5))) == false ? "无" : String.valueOf(row.getField(5)),
                            StringUtils.isNotEmpty(String.valueOf(row.getField(6))) == false ? -1L : Long.valueOf(String.valueOf(row.getField(6))),
                            StringUtils.isNotEmpty(String.valueOf(row.getField(7))) == false ? -1L : Long.valueOf(String.valueOf(row.getField(7))),
                            StringUtils.isNotEmpty(String.valueOf(row.getField(8))) == false ? "无" : String.valueOf(row.getField(8)),
                            StringUtils.isNotEmpty(String.valueOf(row.getField(9))) == false ? "无" : String.valueOf(row.getField(9)),
                            StringUtils.isNotEmpty(String.valueOf(row.getField(10))) == false ? "无" : String.valueOf(row.getField(10)),
                            StringUtils.isNotEmpty(String.valueOf(row.getField(11))) == false ? "无" : String.valueOf(row.getField(11)));
                });
        //extractSessionDetail.print();
        //System.out.println("extractSessionDetail's count : " + extractSessionDetail.count());

        if (ConfigurationManager.getBoolean(Constants.FLINK_LOCAL)) {
            extractSessionDetail.output(
                    JDBCOutputFormat
                            .buildJDBCOutputFormat()
                            .setDrivername(ConfigurationManager.getProperty(Constants.JDBC_DRIVER))
                            .setDBUrl(ConfigurationManager.getProperty(Constants.JDBC_URL))
                            .setUsername(ConfigurationManager.getProperty(Constants.JDBC_USER))
                            .setPassword(ConfigurationManager.getProperty(Constants.JDBC_PASSWORD))
                            .setQuery(ConfigurationManager.getProperty(Constants.SQL_SESSION_DETAIL))
                            .finish()
            );
        } else {
            extractSessionDetail.output(
                    JDBCOutputFormat
                            .buildJDBCOutputFormat()
                            .setDrivername(ConfigurationManager.getProperty(Constants.JDBC_DRIVER))
                            .setDBUrl(ConfigurationManager.getProperty(Constants.JDBC_URL_PROD))
                            .setUsername(ConfigurationManager.getProperty(Constants.JDBC_USER_PROD))
                            .setPassword(ConfigurationManager.getProperty(Constants.JDBC_PASSWORD_PROD))
                            .setQuery(ConfigurationManager.getProperty(Constants.SQL_SESSION_DETAIL))
                            .finish()
            );
        }

    }

    /**
     * 获取top10热门品类
     *
     * @param sessionid2detail
     * @param configuration
     * @return
     * @throws Exception
     */
    private static DataSet<Tuple5<Long, Long, Long, Long, Long>> getTop10Category(DataSet<Tuple2<String, Row>> sessionid2detail, Configuration configuration) throws Exception {
        // 第一步，获取符合条件的session访问过的所有品类id
        // 访问过：指的是，点击过、下单过、支付过的品类
        DataSet<Tuple2<Long, Long>> categoryid = sessionid2detail.flatMap(
                new FlatMapIterator<Tuple2<String, Row>, Tuple2<Long, Long>>() {
                    @Override
                    public Iterator<Tuple2<Long, Long>> flatMap(Tuple2<String, Row> value) throws Exception {
                        Row row = value.f1;
                        List<Tuple2<Long, Long>> list = new ArrayList<>();
                        // clickCategoryId
                        if (StringUtils.isNotEmpty(String.valueOf(row.getField(6)))) {
                            long clickCategoryId = Long.valueOf(String.valueOf(row.getField(6)));
                            list.add(new Tuple2<>(clickCategoryId, clickCategoryId));
                        }

                        // 这里的orderCategoryIds和payCategoryIds可能有多个（假数据中只有一个）
                        if (StringUtils.isNotEmpty(String.valueOf(row.getField(8)))) {
                            String[] orderCategoryIdsSplited = String.valueOf(row.getField(8)).split(",");
                            for (String orderCategoryId : orderCategoryIdsSplited) {
                                list.add(new Tuple2<>(Long.valueOf(orderCategoryId), Long.valueOf(orderCategoryId)));
                            }
                        }
                        if (StringUtils.isNotEmpty(String.valueOf(row.getField(10)))) {
                            String[] payCategoryIdsSplited = String.valueOf(row.getField(10)).split(",");
                            for (String payCategoryId : payCategoryIdsSplited) {
                                list.add(new Tuple2<>(Long.valueOf(payCategoryId), Long.valueOf(payCategoryId)));
                            }
                        }
                        return list.iterator();
                    }
                }
        );
        /**
         * 必须进行去重
         * 如果不去重的话，会出现重复的categoryid，排序会对重复的categoryid以及countInfo进行排序
         * 最后很可能会拿到重复的数据
         */
        categoryid = categoryid.distinct();
        //categoryid.print();
        //System.out.println(categoryid.count());

        // 第二步，计算各品类的点击、下单和支付次数
        // 访问明细中，其中三种访问行为是：点击、下单和支付
        // 分别来计算各品类点击、下单和支付的次数，可以先对访问明细数据进行过滤
        // 分别过滤出点击、下单和支付行为，然后通过各种算子来进行计算

        // 计算各个品类的点击次数
        // TODO 过滤之后会出现数据倾斜，此处应Repartition
        DataSet<Tuple2<Long, Long>> clickCategoryId2Count = sessionid2detail
                .filter((FilterFunction<Tuple2<String, Row>>) value -> {
                    Row row = value.f1;
                    return StringUtils.isNotEmpty(String.valueOf(row.getField(6)));
                })
                .map((MapFunction<Tuple2<String, Row>, Tuple2<Long, Long>>) value -> {
                    Long clickCategoryId = Long.valueOf(String.valueOf(value.f1.getField(6)));
                    return new Tuple2<>(clickCategoryId, 1L);
                })
                .groupBy(0)
                .reduceGroup(new RichGroupReduceFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>() {
                    @Override
                    public void reduce(Iterable<Tuple2<Long, Long>> values, Collector<Tuple2<Long, Long>> out) throws Exception {
                        Long count = 0L;
                        Long clickCategoryId = -1L;
                        for (Tuple2<Long, Long> tuple2 : values) {
                            if (clickCategoryId == -1L) {
                                clickCategoryId = tuple2.f0;
                            }
                            count += tuple2.f1;
                        }
                        out.collect(new Tuple2<>(clickCategoryId, count));
                    }
                });
        //clickCategoryId2Count.print();

        // 计算各个品类的下单次数
        DataSet<Tuple2<Long, Long>> orderCategoryId2Count = sessionid2detail
                .filter((FilterFunction<Tuple2<String, Row>>) value -> {
                    Row row = value.f1;
                    return StringUtils.isNotEmpty(String.valueOf(row.getField(8)));
                })
                .flatMap(new FlatMapIterator<Tuple2<String, Row>, Tuple2<Long, Long>>() {
                    @Override
                    public Iterator<Tuple2<Long, Long>> flatMap(Tuple2<String, Row> value) throws Exception {
                        Row row = value.f1;
                        String orderCategoryIds = String.valueOf(row.getField(8));
                        String[] orderCategoryIdsSplited = orderCategoryIds.split(",");

                        List<Tuple2<Long, Long>> list = new ArrayList<>();

                        for (String orderCategoryId : orderCategoryIdsSplited) {
                            list.add(new Tuple2<>(Long.valueOf(orderCategoryId), 1L));
                        }
                        return list.iterator();
                    }
                })
                .groupBy(0)
                .reduceGroup(new RichGroupReduceFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>() {
                    @Override
                    public void reduce(Iterable<Tuple2<Long, Long>> values, Collector<Tuple2<Long, Long>> out) throws Exception {
                        Long count = 0L;
                        Long orderCategoryId = -1L;
                        for (Tuple2<Long, Long> tuple2 : values) {
                            if (orderCategoryId == -1L) {
                                orderCategoryId = tuple2.f0;
                            }
                            count += tuple2.f1;
                        }
                        out.collect(new Tuple2<>(orderCategoryId, count));
                    }
                });
        //orderCategoryId2Count.print();

        // 计算各个品类的支付次数
        DataSet<Tuple2<Long, Long>> payCategoryId2Count = sessionid2detail
                .filter((FilterFunction<Tuple2<String, Row>>) value -> {
                    Row row = value.f1;
                    return StringUtils.isNotEmpty(String.valueOf(row.getField(10)));
                })
                .flatMap(new FlatMapIterator<Tuple2<String, Row>, Tuple2<Long, Long>>() {
                    @Override
                    public Iterator<Tuple2<Long, Long>> flatMap(Tuple2<String, Row> value) throws Exception {
                        Row row = value.f1;
                        String payCategoryIds = String.valueOf(row.getField(10));
                        String[] payCategoryIdsSplited = payCategoryIds.split(",");

                        List<Tuple2<Long, Long>> list = new ArrayList<>();

                        for (String payCategoryId : payCategoryIdsSplited) {
                            list.add(new Tuple2<>(Long.valueOf(payCategoryId), 1L));
                        }
                        return list.iterator();
                    }
                })
                .groupBy(0)
                .reduceGroup(new RichGroupReduceFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>() {
                    @Override
                    public void reduce(Iterable<Tuple2<Long, Long>> values, Collector<Tuple2<Long, Long>> out) throws Exception {
                        Long count = 0L;
                        Long payCategoryId = -1L;
                        for (Tuple2<Long, Long> tuple2 : values) {
                            if (payCategoryId == -1L) {
                                payCategoryId = tuple2.f0;
                            }
                            count += tuple2.f1;
                        }
                        out.collect(new Tuple2<>(payCategoryId, count));
                    }
                });
        //payCategoryId2Count.print();

        // 第三步，join各品类与它的点击、下单和支付次数
        // 使用 LEFT OUTER JOIN
        DataSet<Tuple2<Long, String>> categoryid2count = categoryid
                .leftOuterJoin(clickCategoryId2Count)
                .where(0)
                .equalTo(0)
                .with((JoinFunction<Tuple2<Long, Long>, Tuple2<Long, Long>, Tuple2<Long, String>>) (first, second) -> {
                    long categoryid1 = first.f0;
                    long clickCount = 0L;

                    if (second != null) {
                        clickCount = second.f1;
                    }

                    String value = Constants.FIELD_CATEGORY_ID + "=" + categoryid1 + "|" +
                            Constants.FIELD_CLICK_COUNT + "=" + clickCount;
                    return new Tuple2<>(categoryid1, value);
                })
                .leftOuterJoin(orderCategoryId2Count)
                .where(0)
                .equalTo(0)
                .with((JoinFunction<Tuple2<Long, String>, Tuple2<Long, Long>, Tuple2<Long, String>>) (first, second) -> {
                    long categoryid12 = first.f0;
                    String value = first.f1;
                    long orderCount = 0L;

                    if (second != null) {
                        orderCount = second.f1;
                    }

                    value = value + "|" + Constants.FIELD_ORDER_COUNT + "=" + orderCount;
                    return new Tuple2<>(categoryid12, value);
                })
                .leftOuterJoin(payCategoryId2Count)
                .where(0)
                .equalTo(0)
                .with((JoinFunction<Tuple2<Long, String>, Tuple2<Long, Long>, Tuple2<Long, String>>) (first, second) -> {
                    long categoryid13 = first.f0;
                    String value = first.f1;
                    long payCount = 0L;

                    if (second != null) {
                        payCount = second.f1;
                    }

                    value = value + "|" + Constants.FIELD_PAY_COUNT + "=" + payCount;
                    return new Tuple2<>(categoryid13, value);
                });
        //categoryid2count.print();
        DataSet<Tuple5<Long, Long, Long, Long, Long>> sortedCategoryCount = categoryid2count
                .map(new RichMapFunction<Tuple2<Long, String>, Tuple5<Long, Long, Long, Long, Long>>() {
                    private long taskid;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        taskid = parameters.getLong("taskid", 0L);
                    }

                    @Override
                    public Tuple5<Long, Long, Long, Long, Long> map(Tuple2<Long, String> value) throws Exception {
                        long categoryid = value.f0;
                        String counts = value.f1;
                        long clickCount = Long.valueOf(StringUtils.getFieldFromConcatString(counts, "\\|", Constants.FIELD_CLICK_COUNT));
                        long orderCount = Long.valueOf(StringUtils.getFieldFromConcatString(counts, "\\|", Constants.FIELD_ORDER_COUNT));
                        long payCount = Long.valueOf(StringUtils.getFieldFromConcatString(counts, "\\|", Constants.FIELD_PAY_COUNT));
                        return new Tuple5<>(taskid, categoryid, clickCount, orderCount, payCount);
                    }
                })
                .withParameters(configuration)
                .sortPartition(2, Order.DESCENDING)
                .sortPartition(3, Order.DESCENDING)
                .sortPartition(4, Order.DESCENDING);

        // TODO sortPartition()
        /**
         *
         * 设置并行度为1的话，性能太差，（并行度只能设一次的话太差）
         * 可不可以将需要排序的数据全部collect到本地进行二次排序？
         *
         * sortPartition是将每个分区中的数据集进行排序
         * 当并行度为1时，排序全局的数据集
         * 而当并行度>1时，只会排序每个分区中的数据，而不是全局的
         */
        //sortedCategoryCount.print();
        //System.out.println(sortedCategoryCount.count());
        //System.out.println(env.getParallelism());

        // 第四步，用first(10)取出top10热门品类，并写入mysql
        DataSet<Tuple5<Long, Long, Long, Long, Long>> top10Category = sortedCategoryCount
                .first(10);
        //top10Category.print();

        if (ConfigurationManager.getBoolean(Constants.FLINK_LOCAL)) {
            top10Category.output(
                    JDBCOutputFormat
                            .buildJDBCOutputFormat()
                            .setDrivername(ConfigurationManager.getProperty(Constants.JDBC_DRIVER))
                            .setDBUrl(ConfigurationManager.getProperty(Constants.JDBC_URL))
                            .setUsername(ConfigurationManager.getProperty(Constants.JDBC_USER))
                            .setPassword(ConfigurationManager.getProperty(Constants.JDBC_PASSWORD))
                            .setQuery(ConfigurationManager.getProperty(Constants.SQL_TOP10_CATEGORY))
                            .finish()
            );
        } else {
            top10Category.output(
                    JDBCOutputFormat
                            .buildJDBCOutputFormat()
                            .setDrivername(ConfigurationManager.getProperty(Constants.JDBC_DRIVER))
                            .setDBUrl(ConfigurationManager.getProperty(Constants.JDBC_URL_PROD))
                            .setUsername(ConfigurationManager.getProperty(Constants.JDBC_USER_PROD))
                            .setPassword(ConfigurationManager.getProperty(Constants.JDBC_PASSWORD_PROD))
                            .setQuery(ConfigurationManager.getProperty(Constants.SQL_TOP10_CATEGORY))
                            .finish()
            );
        }

        return top10Category;
    }

    /**
     * 获取top10活跃session
     *
     * @param sessionid2detail
     * @param top10Category
     */
    private static void getTop10Session(DataSet<Tuple2<String, Row>> sessionid2detail, DataSet<Tuple5<Long, Long, Long, Long, Long>> top10Category) {
        // 第一步，计算各品类被各个session点击的次数
        DataSet<Tuple3<Long, String, Long>> categoryid2sessionCount = sessionid2detail
                .groupBy(0)
                .combineGroup((GroupCombineFunction<Tuple2<String, Row>, Tuple3<Long, String, Long>>) (values, out) -> {
                    String sessionid = "";
                    Map<Long, Long> categoryCountMap = new HashMap<>();
                    Iterator<Tuple2<String, Row>> iterator = values.iterator();

                    // 计算出该session，对每个品类的点击次数
                    while (iterator.hasNext()) {
                        Tuple2<String, Row> tuple2 = iterator.next();
                        if ("".equals(sessionid)) {
                            sessionid = tuple2.f0;
                        }
                        Row row = tuple2.f1;
                        if (StringUtils.isNotEmpty(String.valueOf(row.getField(6)))) {
                            long categoryid = Long.valueOf(String.valueOf(row.getField(6)));

                            // Long long 区别
                            Long count = categoryCountMap.get(categoryid);
                            if (count == null) {
                                count = 0L;
                            }
                            count++;

                            categoryCountMap.put(categoryid, count);
                        }
                    }

                    // 返回结果，<categoryid, sessionid:count>格式
                    for (Map.Entry<Long, Long> categoryCountEntry : categoryCountMap.entrySet()) {
                        long categoryid = categoryCountEntry.getKey();
                        long count = categoryCountEntry.getValue();
                        //String value = sessionid + "," + count;
                        out.collect(new Tuple3<>(categoryid, sessionid, count));
                    }
                });

        // 第二步，获取到top10热门品类，被各个session点击的次数
        DataSet<Tuple4<Long, Long, String, Long>> top10CategorySessionCount = top10Category
                .join(categoryid2sessionCount)
                .where(1)
                .equalTo(0)
                .projectFirst(0)
                .projectFirst(1)
                .projectSecond(1)
                .projectSecond(2);
        //top10CategorySessionCount.print();
        //System.out.println(top10CategorySessionCount.count());

        // 第三步，分组取first 10，获取每个品类的top10活跃用户
        DataSet<Tuple4<Long, Long, String, Long>> top10Session = top10CategorySessionCount
                .groupBy(1)
                .sortGroup(3, Order.DESCENDING)
                .first(10);
        //top10Session.print();
        //System.out.println(top10Session.count());

        // 第四步，将top10Session写入Mysql表
        if (ConfigurationManager.getBoolean(Constants.FLINK_LOCAL)) {
            top10Session.output(
                    JDBCOutputFormat
                            .buildJDBCOutputFormat()
                            .setDrivername(ConfigurationManager.getProperty(Constants.JDBC_DRIVER))
                            .setDBUrl(ConfigurationManager.getProperty(Constants.JDBC_URL))
                            .setUsername(ConfigurationManager.getProperty(Constants.JDBC_USER))
                            .setPassword(ConfigurationManager.getProperty(Constants.JDBC_PASSWORD))
                            .setQuery(ConfigurationManager.getProperty(Constants.SQL_TOP10_CATEGORY_SESSION))
                            .finish()
            );
        } else {
            top10Session.output(
                    JDBCOutputFormat
                            .buildJDBCOutputFormat()
                            .setDrivername(ConfigurationManager.getProperty(Constants.JDBC_DRIVER))
                            .setDBUrl(ConfigurationManager.getProperty(Constants.JDBC_URL_PROD))
                            .setUsername(ConfigurationManager.getProperty(Constants.JDBC_USER_PROD))
                            .setPassword(ConfigurationManager.getProperty(Constants.JDBC_PASSWORD_PROD))
                            .setQuery(ConfigurationManager.getProperty(Constants.SQL_TOP10_CATEGORY_SESSION))
                            .finish()
            );
        }

        // 第五步，获取top10活跃session的明细数据，并写入MySQL
        DataSet<Tuple12<Long, Long, String, Long, String, String, Long, Long, String, String, String, String>> top10SessionDetail = top10Session
                .join(sessionid2detail)
                .where(2)
                .equalTo(0)
                .with((JoinFunction<Tuple4<Long, Long, String, Long>, Tuple2<String, Row>, Tuple12<Long, Long, String, Long, String, String, Long, Long, String, String, String, String>>) (first, second) -> {
                    long taskid = first.f0;
                    Row row = second.f1;
                    String sessionid = first.f2;
                    // 此处返回的数据等会要通过JDBCOutputFormat sink到mysql中，
                    // 所以其中不能出现null值，否则会报空指针异常
                    // Taskid Userid Sessionid Pageid ActionTime SearchKeyword ClickCategoryId ClickProductId OrderCategoryIds OrderProductIds PayCategoryIds PayProductIds
                    return new Tuple12<>(
                            taskid,
                            Long.valueOf(String.valueOf(row.getField(1))),
                            sessionid,
                            Long.valueOf(String.valueOf(row.getField(3))),
                            String.valueOf(row.getField(4)),
                            StringUtils.isNotEmpty(String.valueOf(row.getField(5))) == false ? "无" : String.valueOf(row.getField(5)),
                            StringUtils.isNotEmpty(String.valueOf(row.getField(6))) == false ? -1L : Long.valueOf(String.valueOf(row.getField(6))),
                            StringUtils.isNotEmpty(String.valueOf(row.getField(7))) == false ? -1L : Long.valueOf(String.valueOf(row.getField(7))),
                            StringUtils.isNotEmpty(String.valueOf(row.getField(8))) == false ? "无" : String.valueOf(row.getField(8)),
                            StringUtils.isNotEmpty(String.valueOf(row.getField(9))) == false ? "无" : String.valueOf(row.getField(9)),
                            StringUtils.isNotEmpty(String.valueOf(row.getField(10))) == false ? "无" : String.valueOf(row.getField(10)),
                            StringUtils.isNotEmpty(String.valueOf(row.getField(11))) == false ? "无" : String.valueOf(row.getField(11)));
                });
        //top10SessionDetail.print();
        //System.out.println(top10SessionDetail.count());

        if (ConfigurationManager.getBoolean(Constants.FLINK_LOCAL)) {
            top10SessionDetail.output(
                    JDBCOutputFormat
                            .buildJDBCOutputFormat()
                            .setDrivername(ConfigurationManager.getProperty(Constants.JDBC_DRIVER))
                            .setDBUrl(ConfigurationManager.getProperty(Constants.JDBC_URL))
                            .setUsername(ConfigurationManager.getProperty(Constants.JDBC_USER))
                            .setPassword(ConfigurationManager.getProperty(Constants.JDBC_PASSWORD))
                            .setQuery(ConfigurationManager.getProperty(Constants.SQL_SESSION_DETAIL))
                            .finish()
            );
        } else {
            top10SessionDetail.output(
                    JDBCOutputFormat
                            .buildJDBCOutputFormat()
                            .setDrivername(ConfigurationManager.getProperty(Constants.JDBC_DRIVER))
                            .setDBUrl(ConfigurationManager.getProperty(Constants.JDBC_URL_PROD))
                            .setUsername(ConfigurationManager.getProperty(Constants.JDBC_USER_PROD))
                            .setPassword(ConfigurationManager.getProperty(Constants.JDBC_PASSWORD_PROD))
                            .setQuery(ConfigurationManager.getProperty(Constants.SQL_SESSION_DETAIL))
                            .finish()
            );
        }

    }

    /**
     * 计算各session范围占比，并写入MySQL
     *
     * @param sessionAggrStatMap
     * @param taskid
     */
    private static void calculateAndPersistAggrStat(ConcurrentSkipListMap<String, MutableLong> sessionAggrStatMap, long taskid) {
        // 从Accumulator统计串中获取值
        long session_count = sessionAggrStatMap.get(Constants.SESSION_COUNT).longValue();
        long visit_length_1s_3s = sessionAggrStatMap.get(Constants.TIME_PERIOD_1s_3s).longValue();
        long visit_length_4s_6s = sessionAggrStatMap.get(Constants.TIME_PERIOD_4s_6s).longValue();
        long visit_length_7s_9s = sessionAggrStatMap.get(Constants.TIME_PERIOD_7s_9s).longValue();
        long visit_length_10s_30s = sessionAggrStatMap.get(Constants.TIME_PERIOD_10s_30s).longValue();
        long visit_length_30s_60s = sessionAggrStatMap.get(Constants.TIME_PERIOD_30s_60s).longValue();
        long visit_length_1m_3m = sessionAggrStatMap.get(Constants.TIME_PERIOD_1m_3m).longValue();
        long visit_length_3m_10m = sessionAggrStatMap.get(Constants.TIME_PERIOD_3m_10m).longValue();
        long visit_length_10m_30m = sessionAggrStatMap.get(Constants.TIME_PERIOD_10m_30m).longValue();
        long visit_length_30m = sessionAggrStatMap.get(Constants.TIME_PERIOD_30m).longValue();

        long step_length_1_3 = sessionAggrStatMap.get(Constants.STEP_PERIOD_1_3).longValue();
        long step_length_4_6 = sessionAggrStatMap.get(Constants.STEP_PERIOD_4_6).longValue();
        long step_length_7_9 = sessionAggrStatMap.get(Constants.STEP_PERIOD_7_9).longValue();
        long step_length_10_30 = sessionAggrStatMap.get(Constants.STEP_PERIOD_10_30).longValue();
        long step_length_30_60 = sessionAggrStatMap.get(Constants.STEP_PERIOD_30_60).longValue();
        long step_length_60 = sessionAggrStatMap.get(Constants.STEP_PERIOD_60).longValue();

        // 计算各个访问时长和访问步长的范围
        double visit_length_1s_3s_ratio = NumberUtils.formatDouble(
                (double) visit_length_1s_3s / (double) session_count, 2);
        double visit_length_4s_6s_ratio = NumberUtils.formatDouble(
                (double) visit_length_4s_6s / (double) session_count, 2);
        double visit_length_7s_9s_ratio = NumberUtils.formatDouble(
                (double) visit_length_7s_9s / (double) session_count, 2);
        double visit_length_10s_30s_ratio = NumberUtils.formatDouble(
                (double) visit_length_10s_30s / (double) session_count, 2);
        double visit_length_30s_60s_ratio = NumberUtils.formatDouble(
                (double) visit_length_30s_60s / (double) session_count, 2);
        double visit_length_1m_3m_ratio = NumberUtils.formatDouble(
                (double) visit_length_1m_3m / (double) session_count, 2);
        double visit_length_3m_10m_ratio = NumberUtils.formatDouble(
                (double) visit_length_3m_10m / (double) session_count, 2);
        double visit_length_10m_30m_ratio = NumberUtils.formatDouble(
                (double) visit_length_10m_30m / (double) session_count, 2);
        double visit_length_30m_ratio = NumberUtils.formatDouble(
                (double) visit_length_30m / (double) session_count, 2);

        double step_length_1_3_ratio = NumberUtils.formatDouble(
                (double) step_length_1_3 / (double) session_count, 2);
        double step_length_4_6_ratio = NumberUtils.formatDouble(
                (double) step_length_4_6 / (double) session_count, 2);
        double step_length_7_9_ratio = NumberUtils.formatDouble(
                (double) step_length_7_9 / (double) session_count, 2);
        double step_length_10_30_ratio = NumberUtils.formatDouble(
                (double) step_length_10_30 / (double) session_count, 2);
        double step_length_30_60_ratio = NumberUtils.formatDouble(
                (double) step_length_30_60 / (double) session_count, 2);
        double step_length_60_ratio = NumberUtils.formatDouble(
                (double) step_length_60 / (double) session_count, 2);

        // 将统计结果封装成Domain对象
        SessionAggrStat sessionAggrStat = new SessionAggrStat();
        sessionAggrStat.setTaskid(taskid);
        sessionAggrStat.setSession_count(session_count);
        sessionAggrStat.setVisit_length_1s_3s_ratio(visit_length_1s_3s_ratio);
        sessionAggrStat.setVisit_length_4s_6s_ratio(visit_length_4s_6s_ratio);
        sessionAggrStat.setVisit_length_7s_9s_ratio(visit_length_7s_9s_ratio);
        sessionAggrStat.setVisit_length_10s_30s_ratio(visit_length_10s_30s_ratio);
        sessionAggrStat.setVisit_length_30s_60s_ratio(visit_length_30s_60s_ratio);
        sessionAggrStat.setVisit_length_1m_3m_ratio(visit_length_1m_3m_ratio);
        sessionAggrStat.setVisit_length_3m_10m_ratio(visit_length_3m_10m_ratio);
        sessionAggrStat.setVisit_length_10m_30m_ratio(visit_length_10m_30m_ratio);
        sessionAggrStat.setVisit_length_30m_ratio(visit_length_30m_ratio);
        sessionAggrStat.setStep_length_1_3_ratio(step_length_1_3_ratio);
        sessionAggrStat.setStep_length_4_6_ratio(step_length_4_6_ratio);
        sessionAggrStat.setStep_length_7_9_ratio(step_length_7_9_ratio);
        sessionAggrStat.setStep_length_10_30_ratio(step_length_10_30_ratio);
        sessionAggrStat.setStep_length_30_60_ratio(step_length_30_60_ratio);
        sessionAggrStat.setStep_length_60_ratio(step_length_60_ratio);

        // 调用对应的DAO插入统计结果
        ISessionAggrStatDAO sessionAggrStatDAO = DAOFactory.getSessionAggrStatDAO();
        sessionAggrStatDAO.insert(sessionAggrStat);
    }
}
